package com.example.mq.mqclient;

import com.example.mq.common.BinaryTool;
import com.example.mq.common.Consumer;
import com.example.mq.common.MqException;
import com.example.mq.common.communication.*;
import com.example.mq.mqserver.core.BasicProperties;
import com.example.mq.mqserver.core.ExchangeType;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import javax.xml.stream.events.DTD;
import java.io.IOException;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
@Data
public class Channel {
    private String channelId;
    //标识当前channel属于哪个连接
    private Connection connection;
    //维护一个哈希表，用来存储服务器返回来的响应
    //key-rid
    private ConcurrentHashMap<String,BasicReturns> basicReturnsMap = new ConcurrentHashMap<>();
    //如果某个channel订阅了消息，那么就需要服务器在推送回来消息的时候，执行一下回调函数，也就是消费推送回来的消息
    //这里约定一个channel中只能有一个consumer来执行回调
    private Consumer consumer;

    public Channel(String channelId,Connection connection) {
        this.connection = connection;
        this.channelId = channelId;
    }
    //与服务器之间的交互
    //通过这个类告诉服务器我要创建一个channel
    public boolean createChannel() throws IOException {
        //1.构造payload
        BasicArguments basicArguments = new BasicAckArguments();
        basicArguments.setRid(createNewRid());
        basicArguments.setChannelId(channelId);
        byte[] payload = BinaryTool.toBytes(basicArguments);
        //2.构造响应
        Request request = new Request();
        request.setPayload(payload);
        request.setLength(payload.length);
        request.setType(0x1);

        //3.发送请求
        connection.writeRequest(request);
        //4.等待服务器的响应
        BasicReturns basicReturns = waitBasicReturns(basicArguments.getRid());
        return basicReturns.isOk();
    }
    private BasicReturns waitBasicReturns(String rid) {
        BasicReturns basicReturns = null;
        while((basicReturns = basicReturnsMap.get(rid)) == null) {
            synchronized (this) {
                try {
                    wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                    log.info("[Channel] 服务器响应返回，等待结束");
                }
            }
        }
        //在返回结果之前还需要把对应哈希表中的数据给进行清除
        //虽然每一次生成的rid都不同，好像也不会影响后续连接但是这个记录会一直在哈希表中，直到服务器停止
        //但是可以让内存压力小一点
        basicReturnsMap.remove(rid);
        return basicReturns;
    }

    private String createNewRid() {
        return "R-" + UUID.randomUUID();
    }

    //在Connection中扫描到了对应的响应在调用这个方法
    public void putReturns(BasicReturns basicReturns) {
        basicReturnsMap.put(basicReturns.getRid(),basicReturns);
        synchronized (this) {
            //由于当前不知道等待的线程有多少，所以直接全部唤醒
            notifyAll();
        }
    }

    //告诉服务器销毁channel
    public boolean closeChannel() throws IOException {
        //1.构造payload
        BasicArguments basicArguments = new BasicAckArguments();
        basicArguments.setChannelId(channelId);
        basicArguments.setRid(createNewRid());
        byte[] payload = BinaryTool.toBytes(basicArguments);
        //2.构造响应
        Request request = new Request();
        request.setPayload(payload);
        request.setLength(payload.length);
        request.setType(0x2);
    //3.发送请求
        connection.writeRequest(request);
    //4.等待服务器的响应
    BasicReturns basicReturns = waitBasicReturns(basicArguments.getRid());
        return basicReturns.isOk();
}

    //告诉服务器创建一个交换机
    public boolean exchangeDeclare(String exchangeName, ExchangeType type, boolean durable, boolean autoDelete,
                                   Map<String,Object> arguments) throws IOException {
        //1.构造payload
        ExchangeDeclareArguments exchangeDeclareArguments = new ExchangeDeclareArguments();
        exchangeDeclareArguments.setExchangeName(exchangeName);
        exchangeDeclareArguments.setType(type);
        exchangeDeclareArguments.setDurable(durable);
        exchangeDeclareArguments.setAutoDelete(autoDelete);
        exchangeDeclareArguments.setArguments(arguments);
        exchangeDeclareArguments.setChannelId(channelId);
        exchangeDeclareArguments.setRid(createNewRid());
        byte[] payload = BinaryTool.toBytes(exchangeDeclareArguments);
        //2.构造响应
        Request request = new Request();
        request.setPayload(payload);
        request.setLength(payload.length);
        request.setType(0x3);
        //3.发送请求
        connection.writeRequest(request);
        //4.等待服务器的响应
        BasicReturns basicReturns = waitBasicReturns(exchangeDeclareArguments.getRid());
        return basicReturns.isOk();
    }
    //告诉服务器销毁交换机
    public boolean exchangeDelete(String exchangeName) throws IOException {
        //1.构造payload
        ExchangeDeleteArguments exchangeDeleteArguments = new ExchangeDeleteArguments();
        exchangeDeleteArguments.setExchangeName(exchangeName);
        exchangeDeleteArguments.setChannelId(channelId);
        exchangeDeleteArguments.setRid(createNewRid());
        byte[] payload = BinaryTool.toBytes(exchangeDeleteArguments);
        //2.构造响应
        Request request = new Request();
        request.setPayload(payload);
        request.setLength(payload.length);
        request.setType(0x4);
        //3.发送请求
        connection.writeRequest(request);
        //4.等待服务器的响应
        BasicReturns basicReturns = waitBasicReturns(exchangeDeleteArguments.getRid());
        return basicReturns.isOk();
    }

    //告诉服务器创建一个队列
    public boolean queueDeclare(String queueName,boolean durable,boolean autoDelete,boolean exclusive,
                             Map<String,Object> arguments) throws IOException {
        //1.构造payload
        QueueDeclareArguments queueDeclareArguments = new QueueDeclareArguments();
        queueDeclareArguments.setQueueName(queueName);
        queueDeclareArguments.setDurable(durable);
        queueDeclareArguments.setExclusive(exclusive);
        queueDeclareArguments.setAutoDelete(autoDelete);
        queueDeclareArguments.setRid(createNewRid());
        queueDeclareArguments.setArguments(arguments);
        queueDeclareArguments.setChannelId(channelId);
        byte[] payload = BinaryTool.toBytes(queueDeclareArguments);
        //2.构造响应
        Request request = new Request();
        request.setPayload(payload);
        request.setLength(payload.length);
        request.setType(0x5);
        //3.发送请求
        connection.writeRequest(request);
        //4.等待服务器的响应
        BasicReturns basicReturns = waitBasicReturns(queueDeclareArguments.getRid());
        return basicReturns.isOk();
    }
    //告诉服务器销毁队列
    public boolean queueDelete(String queueName) throws IOException {
        //1.构造payload
        QueueDeleteArguments queueDeleteArguments = new QueueDeleteArguments();
        queueDeleteArguments.setQueueName(queueName);
        queueDeleteArguments.setChannelId(channelId);
        queueDeleteArguments.setRid(createNewRid());
        byte[] payload = BinaryTool.toBytes(queueDeleteArguments);
        //2.构造响应
        Request request = new Request();
        request.setPayload(payload);
        request.setLength(payload.length);
        request.setType(0x6);
        //3.发送请求
        connection.writeRequest(request);
        //4.等待服务器的响应
        BasicReturns basicReturns = waitBasicReturns(queueDeleteArguments.getRid());
        return basicReturns.isOk();
    }
    //告诉服务器创建一个绑定
    public boolean queueBind(String queueName,String exchangeName,String bindingKey) throws IOException {
        //1.构造payload
        QueueBindArguments queueBindArguments = new QueueBindArguments();
        queueBindArguments.setBindingKey(bindingKey);
        queueBindArguments.setQueueName(queueName);
        queueBindArguments.setExchangeName(exchangeName);
        queueBindArguments.setRid(createNewRid());
        queueBindArguments.setChannelId(channelId);
        byte[] payload = BinaryTool.toBytes(queueBindArguments);
        //2.构造响应
        Request request = new Request();
        request.setPayload(payload);
        request.setLength(payload.length);
        request.setType(0x7);
        //3.发送请求
        connection.writeRequest(request);
        //4.等待服务器的响应
        BasicReturns basicReturns = waitBasicReturns(queueBindArguments.getRid());
        return basicReturns.isOk();
    }
    //告诉服务器销毁一个绑定
    public boolean queueUnBind(String queueName,String exchangeName) throws IOException {
        //1.构造payload
        QueueUnBindArguments queueUnBindArguments = new QueueUnBindArguments();
        queueUnBindArguments.setQueueName(queueName);
        queueUnBindArguments.setExchangeName(exchangeName);
        queueUnBindArguments.setRid(createNewRid());
        queueUnBindArguments.setChannelId(channelId);
        byte[] payload = BinaryTool.toBytes(queueUnBindArguments);
        //2.构造响应
        Request request = new Request();
        request.setPayload(payload);
        request.setLength(payload.length);
        request.setType(0x8);
        //3.发送请求
        connection.writeRequest(request);
        //4.等待服务器的响应
        BasicReturns basicReturns = waitBasicReturns(queueUnBindArguments.getRid());
        return basicReturns.isOk();
    }
    //告诉服务器发送一个消息
    public boolean basicPublish(String exchangeName, String routingKey, BasicProperties basicProperties, byte[] body) throws IOException {
        //1.构造payload
        BasicPublishArguments basicPublishArguments = new BasicPublishArguments();
        basicPublishArguments.setBasicProperties(basicProperties);
        basicPublishArguments.setBody(body);
        basicPublishArguments.setRoutingKey(routingKey);
        basicPublishArguments.setChannelId(channelId);
        basicPublishArguments.setExchangeName(exchangeName);
        basicPublishArguments.setRid(createNewRid());
        byte[] payload = BinaryTool.toBytes(basicPublishArguments);
        //2.构造响应
        Request request = new Request();
        request.setPayload(payload);
        request.setLength(payload.length);
        request.setType(0x9);
        //3.发送请求
        connection.writeRequest(request);
        //4.等待服务器的响应
        BasicReturns basicReturns = waitBasicReturns(basicPublishArguments.getRid());
        return basicReturns.isOk();
    }
    //告诉服务器某个消费者要订阅消息
    public boolean basicConsumer(String queueName, boolean autoAck, Consumer consumer) throws IOException {
        //首先设置一下回调
        if(this.consumer != null) {
            throw new IOException("[Channel] 当前channel中已经设置过回调，不能重复设置");
        }
        this.consumer = consumer;
        //1.构造payload
        BasicConsumerArguments basicConsumerArguments = new BasicConsumerArguments();
        //直接用当前channelId代表消费者
        basicConsumerArguments.setConsumerTag(channelId);
        basicConsumerArguments.setRid(createNewRid());
        basicConsumerArguments.setAutoAck(autoAck);
        basicConsumerArguments.setChannelId(channelId);
        basicConsumerArguments.setQueueName(queueName);
        byte[] payload = BinaryTool.toBytes(basicConsumerArguments);
        //2.构造响应
        Request request = new Request();
        request.setPayload(payload);
        request.setLength(payload.length);
        request.setType(0xa);
        //3.发送请求
        connection.writeRequest(request);
        //4.等待服务器的响应
        BasicReturns basicReturns = waitBasicReturns(basicConsumerArguments.getRid());
        return basicReturns.isOk();
    }
    //告诉服务器手动确认
    public boolean basicAck(String queueName,String messageId) throws IOException {
        //1.构造payload
        BasicAckArguments basicAckArguments = new BasicAckArguments();
        basicAckArguments.setMessageId(messageId);
        basicAckArguments.setQueueName(queueName);
        basicAckArguments.setRid(createNewRid());
        basicAckArguments.setChannelId(channelId);
        byte[] payload = BinaryTool.toBytes(basicAckArguments);
        //2.构造响应
        Request request = new Request();
        request.setPayload(payload);
        request.setLength(payload.length);
        request.setType(0xb);
        //3.发送请求
        connection.writeRequest(request);
        //4.等待服务器的响应
        BasicReturns basicReturns = waitBasicReturns(basicAckArguments.getRid());
        return basicReturns.isOk();
    }

}
